package com.song.hbasechap_7_6;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Mapper;

public class IndexMapper  extends TableMapper<Writable, Writable> {

	
	// col:test2;col:test3
	
	private static String cols ="col_test2;col_test3";
	
	
	private Text k = new Text();
	private Text v = new Text();
	
	@Override
	protected void map(
			ImmutableBytesWritable row,
			Result columns,
			Mapper<ImmutableBytesWritable, Result, Writable, Writable>.Context context)
			throws IOException, InterruptedException {
	
		
		String rowkey = new String( row.get());
		
		List<String> seg = Arrays.asList(cols.split(";")) ;
		Map<String,String> colmap = new HashMap<String, String>();
		List<String> keylist = new ArrayList<String>(); 
		List<KeyValue> kvs = columns.list();
		for(KeyValue kv :kvs)
		{
			String family = new String(kv.getFamily());
			String qualifiy = new String(kv.getQualifier());
			String val = new String(kv.getValue());
			String ts = kv.getTimestamp()+"";
			String keystr =family+"_"+qualifiy;
			if(seg.contains(keystr))
			{
				keystr = keystr+"_"+val+"_"+ts;
				keylist.add(keystr);
			}
		}
		if(keylist.size()>0)
		{
			
			String keystr =keylist.toString().replaceAll(",", "_").replaceAll("\\[", "").replaceAll("\\]", "");
			String valstr  = rowkey;
			k.set(Bytes.toBytes(keystr));
			v.set(Bytes.toBytes(valstr));
			context.write( k,v);
		}
	}

	
	
	
}
